package com.swak.reactivex.threads.waitstrategy;

import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import com.swak.reactivex.threads.Barrier;
import com.swak.reactivex.threads.WaitStrategy;

/**
 * SleepingWaitStrategy. 但是可以唤醒.
 * 
 * @author 618lf
 */
public class WakeupWaitStrategy implements WaitStrategy {

	private static final int DEFAULT_RETRIES = 200;
	private static final long DEFAULT_SLEEP = 100;

	private final int retries;
	private final long sleepTimeNs;

	// wait
	private final Lock lock = new ReentrantLock();
	private final Condition processorNotifyCondition = lock.newCondition();

	public WakeupWaitStrategy() {
		this(DEFAULT_RETRIES, DEFAULT_SLEEP);
	}

	public WakeupWaitStrategy(int retries) {
		this(retries, DEFAULT_SLEEP);
	}

	public WakeupWaitStrategy(int retries, long sleepTimeNs) {
		this.retries = retries;
		this.sleepTimeNs = sleepTimeNs;
	}

	@Override
	public void waitFor(Barrier barrier) throws InterruptedException, TimeoutException {
		int counter = retries;
		long timeout = Long.MAX_VALUE;
		while ((timeout = barrier.check()) != 0) {
			counter = applyWaitMethod(counter, timeout);
		}
	}

	private int applyWaitMethod(int counter, long timeout) throws InterruptedException {
		if (counter > 100) {
			--counter;
		} else if (counter > 0) {
			--counter;
			Thread.yield();
		} else {
			lock.lock();
			try {
				processorNotifyCondition.awaitNanos(Math.min(sleepTimeNs, timeout));
			} finally {
				lock.unlock();
			}
		}

		return counter;
	}

	@Override
	public void wakeup() {
		lock.lock();
		try {
			processorNotifyCondition.signalAll();
		} finally {
			lock.unlock();
		}
	}
}
